/*
 * MIT License
 *
 * Copyright (c) 2020 wen.gu <454727014@qq.com>
 *
 * Permission is hereby granted, free of charge, to any person obtaining a copy
 * of this software and associated documentation files (the "Software"), to deal
 * in the Software without restriction, including without limitation the rights
 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
 * copies of the Software, and to permit persons to whom the Software is
 * furnished to do so, subject to the following conditions:
 *
 * The above copyright notice and this permission notice shall be included in all
 * copies or substantial portions of the Software.
 *
 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
 * SOFTWARE.
 */

 /***************************************************************************
 * Name: uds_transport.c
 *
 * Purpose: implementation implementation unix domain socket transport(dialer and listener) ops for cbus
 *
 * Developer:
 *   wen.gu , 2022-02-16
 *
 * TODO:
 *
 ***************************************************************************/

 /******************************************************************************
 **    INCLUDES
 ******************************************************************************/

#if defined(_MSC_VER)
#include "uds_transport.h"

#include "cbus/core/cbus_worker.h"
#include "cbus/core/cbus_map.h"
#include "cbus/core/cbus_string.h"

#include "cbus/core/cbus_log.h"
/******************************************************************************
 **    MACROS
 ******************************************************************************/
#define LISTENER_RECEIVE_BUF_SIZE (1500)
/******************************************************************************
 **    VARIABLE DEFINITIONS
 ******************************************************************************/

struct _uds_listener;
typedef struct _uds_listener uds_listener_t;

typedef struct _listener_client
{
    cbus_dialer_id dialer_id;
    GSTR client_url;    
    cbus_worker_t receive_worker;
    uds_listener_t* listener_ref;
    void* opaque;
    cbus_listener_receive_handler on_receive;
}listener_client_t;

struct _uds_listener
{
    socket_t socket_;
    cbus_worker_t on_client_worker_;
    void* status_opaque;
    cbus_on_dialer_status_func on_dialer_status;

    void *receive_opaque;
    cbus_listener_receive_handler on_receive;

    cbus_map_t client_map;
};
/******************************************************************************
 **    inner FUNCTION DEFINITIONS
 ******************************************************************************/

static void listener_notify_status(uds_listener_t* listener, cbus_dialer_id client_id, GBOL is_connected)
{
    if (listener->on_dialer_status)
    {
        listener->on_dialer_status(listener->status_opaque, client_id, is_connected);
    }
}


static void listener_client_del(uds_listener_t* listener, GS32 client_id);

static void listener_on_receive_worker(cbus_worker_t thd_worker, void* opaque)
{
    listener_client_t* client = (listener_client_t*)opaque;
    fd_set read_set;
    int max_fd = -1;
    int ret = 0;
    struct timeval timeout;  //timeout time    
    timeout.tv_sec = SELECT_TIME_OUT_SEC;   /** */
    timeout.tv_usec = 0;   
    socket_t socket_id = (socket_t)(client->dialer_id); 
    char receive_buf[LISTENER_RECEIVE_BUF_SIZE] = {0};

    LOGD("[%s.%d]===>:\n", __FUNCTION__, __LINE__);

    while (cbus_worker_running(thd_worker))
    {
        FD_ZERO(&read_set);
        
        FD_SET(socket_id, &read_set);
        max_fd = socket_id;
        ret = select(max_fd + 1, &read_set, NULL, NULL, &timeout);
        
        if ((ret > 0) && FD_ISSET(socket_id, &read_set))
        {        
            int len = recv(socket_id, receive_buf, 1500, 0);
            if (len > 0)
            {
                if (client->listener_ref->on_receive)
                {
                    client->listener_ref->on_receive(client->listener_ref->receive_opaque, (cbus_dialer_id)socket_id, (GBUF)receive_buf, len);
                }
                else if (client->on_receive)
                {
                    client->on_receive(client->opaque, (cbus_dialer_id)socket_id, (GBUF)receive_buf, len);
                }
            }
            else if (len == 0)
            {
                listener_client_del(client->listener_ref, socket_id);
            }
            else
            {
                if (errno != EINTR)
                {
                    listener_client_del(client->listener_ref, socket_id);
                }
                else
                {
                    LOGE("read error: %s\n", strerror(errno));
                }
            }            
        }
    }
}


static listener_client_t* listener_client_create(uds_listener_t* listener, cbus_dialer_id dialer_id, const GSTR client_url)
{
    listener_client_t* client = (listener_client_t*)malloc(sizeof(listener_client_t));
    if (client)
    {        
        GErrc ret = cbus_worker_create(&client->receive_worker, client_url, listener_on_receive_worker, client);

        if (G_OK != ret)
        {
            free(client);
            return NULL;
        }

        client->listener_ref = listener;
        client->dialer_id = dialer_id;
        client->client_url = cbus_str_dup(client_url);
    }    

    return client;
}

static void listener_client_free(void* ptr)
{
    if (ptr)
    {
        listener_client_t* client = (listener_client_t*)ptr;
        if (client->receive_worker)
        {
            cbus_worker_join(client->receive_worker);
        }
        
        closesocket((socket_t)client->dialer_id);
        free(ptr);
    }
}

static GErrc listener_client_start(listener_client_t* client)
{
    return cbus_worker_start(client->receive_worker);
}

#if 0
static GErrc listener_client_pause(listener_client_t* client)
{
    return cbus_worker_pause(client->receive_worker);
}
#endif

//////////////////////////////////////////////////////////////////////////////////////////////////
//////////////////////////////////////////////////////////////////////////////////////////////////
static GErrc listener_socket_create_and_bind(uds_listener_t* listener, const GSTR url)
{
  if (listener->socket_ != INVALID_SOCKET)
    {
        LOGE("serivce socket already exist, please destroy old one before create operation\n");
        return G_ErrInvalidStatus;
    }
    if ((listener->socket_ = socket(AF_UNIX, SOCK_STREAM, 0)) < 0)
    {
        LOGE("create AF_UNIX socket failed(%ld)\n", WSAGetLastError());
        listener->socket_ = INVALID_SOCKET;
        return G_ErrInsufficientResources;
    }

    struct sockaddr_un srv_addr;
    memset(&srv_addr, 0, sizeof(srv_addr));            /* fill socket address structure with our address */
    srv_addr.sun_family = AF_UNIX;

    unlink(url);

    snprintf(srv_addr.sun_path, sizeof(srv_addr.sun_path), "%s", url);

    LOGI("service addr:%s, len:%d\n", srv_addr.sun_path, sizeof(srv_addr));

    int32_t ret = bind(listener->socket_, (struct sockaddr*) &srv_addr, sizeof(srv_addr));
    if (ret == SOCKET_ERROR)
    {
        LOGE("bind service socket:%s, failed: %ld\n", srv_addr.sun_path, WSAGetLastError());
        closesocket(listener->socket_);
        listener->socket_ = INVALID_SOCKET;
        return G_ErrUndefined;
    }

    ret = listen(listener->socket_, MAX_CLIENT_CONNECT_LIMIT);

    if (ret < 0)
    {
        LOGE("listen socket:%s, failed:%ld\n", srv_addr.sun_path, WSAGetLastError());
        closesocket(listener->socket_);
        listener->socket_ = INVALID_SOCKET;
        return G_ErrUndefined;
    }

    return G_OK;
}

static GErrc listener_socket_destroy(uds_listener_t* listener)
{
    if (listener->socket_ != INVALID_SOCKET)
    {
        closesocket(listener->socket_);
        listener->socket_ = INVALID_SOCKET;
    }

    return G_OK;
}

static void listener_client_add(uds_listener_t* listener, cbus_dialer_id client_id, const GSTR client_url)
{
    listener_client_t* client = listener_client_create(listener, client_id, client_url);
    LOGD("[%s.%d]===>: receive client(%s) connect request\n", __FUNCTION__, __LINE__, client_url);
    if (client)
    {
        if (cbus_map_set_pointer(listener->client_map, (GU64)client_id, client, listener_client_free) == GTRUE)
        {
            listener_notify_status(listener, client_id, GTRUE);
            GErrc ret = listener_client_start(client);

            if (G_OK != ret)
            {
                LOGE("start receive worker for client(%s) failed\n", client_url); //todo proccess the error
            }
        }
        else
        {
            listener_client_free(client);
        }
    }
}

static void listener_client_del(uds_listener_t* listener, GS32 client_id)
{
    cbus_map_remove(listener->client_map, client_id);
}



static void listener_on_client_connect_worker(cbus_worker_t thd_worker, void* opaque)
{
    uds_listener_t* listener = (uds_listener_t*)opaque;
    struct sockaddr_un un_client;
    int len = sizeof(un_client);
    socket_t cli_fd = -1;
    uint16_t client_id = 0;		
    
    fd_set write_set;
    int max_fd = -1;
    int ret = 0;
    struct timeval timeout;  //timeout time    
    timeout.tv_sec = SELECT_TIME_OUT_SEC;   /** */
    timeout.tv_usec = 0;    
    LOGD("[%s.%d]===>:\n", __FUNCTION__, __LINE__);
    while (cbus_worker_running(thd_worker))
    {
        memset(un_client.sun_path, 0, sizeof(un_client.sun_path));
        
        max_fd = -1;
        FD_ZERO(&write_set);
        
        FD_SET(listener->socket_, &write_set);
        max_fd = listener->socket_;
        
        ret = select(max_fd + 1, NULL, &write_set, NULL, &timeout);
        
        if (ret > 0 && FD_ISSET(listener->socket_, &write_set))
        {
            if ((cli_fd = accept(listener->socket_, (struct sockaddr*)&un_client, &len)) <= 0)
            {
                continue;
            }
        
            listener_client_add(listener, (cbus_dialer_id)cli_fd, un_client.sun_path);
        }
    }
}

#if 0
static void listener_client_on_start(void* opaque, GU64 key, GU32 type, const void* value, GU32 value_size)
{
    listener_client_t* client = (listener_client_t*)value;
    GErrc ret = cbus_worker_start(client->receive_worker);

    if (G_OK != ret)
    {
        LOGE("start receive worker for(%s) failed\n", client->client_url);
    }
}
#endif

static GErrc listener_worker_start(uds_listener_t* listener)
{
    LOGD("[%s.%d]===>:\n", __FUNCTION__, __LINE__);
    return cbus_worker_start(listener->on_client_worker_);
}

static GErrc listener_worker_stop(uds_listener_t* listener)
{
    cbus_worker_pause(listener->on_client_worker_);
    cbus_map_clear(listener->client_map);
    return G_OK;
}

/******************************************************************************
 **    FUNCTION DEFINITIONS
 ******************************************************************************/
/**
 * @brief create and initialize a listener instance
 * @param handle [out] the instance of listener(service)
 * @param opaque [in] a oapque handle of user set to listener for callback function
 * @param on_dialer_status [in] a callback to observe the status of dialer which connec/disconnect
 *                          to/from current listener
 * @return GErrc, success: G_OK, else error code.
 * 
 */
GErrc uds_listener_initialize(GPHD* handle, cbus_on_dialer_status_func on_dialer_satus, void* opaque)
{
    if (!handle)
    {
        return G_ErrBadParameter;
    }

    uds_listener_t* listener = (uds_listener_t*)malloc(sizeof(uds_listener_t));

    if (listener)
    {
        memset(listener, 0, sizeof(uds_listener_t));
        listener->client_map = cbus_map_create(CMV_TYPE_Pointer, CBUS_MAP_DEFAULT_CAPACITY);
        if (!listener->client_map)
        {
            free(listener);
            return G_ErrInsufficientResources;
        }
        LOGD("[%s.%d]===>:\n", __FUNCTION__, __LINE__);
        
        GErrc ret = cbus_worker_create(&listener->on_client_worker_, "on_dialer_status", listener_on_client_connect_worker, listener);
        LOGD("[%s.%d]===>:\n", __FUNCTION__, __LINE__);
        if (G_OK != ret)
        {
            cbus_map_destroy(listener->client_map);
            free(listener);
            return G_ErrInsufficientResources;
        }

        listener->on_dialer_status = on_dialer_satus;
        listener->status_opaque = opaque;
        listener->socket_ = INVALID_SOCKET;
        *handle = listener;
        return G_OK;
    }

    return G_ErrInsufficientResources;
}

/**
 * @brief destroy a listener instance 
 * @param handle [in] the instance handle of listener
 * @return GErrc, success: G_OK, else error code.
 *  
 */
GErrc uds_listener_finish(GPHD handle)
{
    if (!handle)
    {
        return G_ErrBadParameter;
    }

    uds_listener_t* listener = (uds_listener_t*)handle;

    if (listener->on_client_worker_)
    {
        cbus_worker_join(listener->on_client_worker_);
    }

    if (listener->client_map)
    {
        cbus_map_destroy(listener->client_map);
    }

    free(listener);

    return G_OK;
}

/**
 * @brief offer a listener(service) with url address 
 * @param handle [in] the instance handle of listener
 * @param url [in] the url address of a listener(service)
 * @return GErrc, success: G_OK, else error code.
 */
GErrc uds_listener_bind(GPHD handle, GSTR url)
{
    if (!handle || !url || (strlen(url) <= UDS_TRANSPORT_URL_PREFIX_LEN)) /** url len must > UDS_TRANSPORT_URL_PREFIX_LEN */
    {
        return G_ErrBadParameter;
    }

    uds_listener_t* listener = (uds_listener_t*)handle;

    GErrc ret = listener_socket_create_and_bind(listener, url + UDS_TRANSPORT_URL_PREFIX_LEN);
    LOGD("[%s.%d]===>:\n", __FUNCTION__, __LINE__);
    if (G_OK == ret)
    {
        LOGD("[%s.%d]===>:\n", __FUNCTION__, __LINE__);
        ret = listener_worker_start(listener);
        LOGD("[%s.%d]===>:\n", __FUNCTION__, __LINE__);
    }

    return ret;
}

/**
 * @brief stop a listner(service)
 * @param handle [in] the instance handle of listener
 * @return GErrc, success: G_OK, else error code.
 */
GErrc uds_listener_unbind(GPHD handle)
{
    if (!handle)
    {
        return G_ErrBadParameter;
    }

    uds_listener_t* listener = (uds_listener_t*)handle;

    GErrc ret = listener_worker_stop(listener);

    if (G_OK == ret)
    {
        ret = listener_socket_destroy(listener);
    }

    return ret;
}

/**
 * @brief send data to remote side
 * @param handle [in] the instance handle of listener
 * @param buf [in] data buffer
 * @param len [in] data length
 * @return G_OK: success, else: failed @see GErrc 
 * 
 */
GErrc uds_listener_send(GPHD handle, cbus_dialer_id dialer_id, GBUF buf, GU32 len)
{
    if (!handle || (((socket_t)dialer_id) == INVALID_SOCKET))
    {
        return G_ErrBadParameter;
    }

    uds_listener_t* listener = (uds_listener_t*)handle;

    while (len > 0)
    {
        int32_t ret_len = send((socket_t)dialer_id, (char*)buf, len, 0);
        if (ret_len > 0)
        {
            len -= ret_len;
            buf += ret_len;
        }
        else if (ret_len == 0)
        {
            listener_notify_status(listener, dialer_id, GFALSE);
            return G_ErrDisconnected;
        }
        else
        {
            if (errno != EINTR)
            {
                listener_notify_status(listener, dialer_id, GFALSE); // todo refine me??
                return G_ErrDisconnected;
            }
            /** todo something */
        }
    } 

    return G_OK;
}

/**
 * @brief set a receiver handler as callback to receive data from remote side
 * @param handle [in] the instance handle of listener
 * @param opaque [in] a oapque handle of user for Receiver handler
 * @param on_receive [in] a callback for receive data from remote side
 * @return success:G_OK, else: error code @see GErrc
 * 
 */
GErrc uds_listener_set_receiver(GPHD handle, cbus_dialer_id dialer_id, cbus_listener_receive_handler on_receive, void* opaque)
{
    if (!handle || !on_receive)
    {
        return G_ErrBadParameter;
    }

    uds_listener_t* listener = (uds_listener_t*)handle;

    if (dialer_id == CBUS_INVALID_DIALER_ID)
    {
        listener->on_receive = on_receive;
        listener->receive_opaque = opaque;        
    }
    else
    {
        if (!listener->client_map)
        {
            return G_ErrInvalidStatus;
        }

        listener_client_t* client = NULL;
        if (cbus_map_find_pointer(listener->client_map, (GU64)dialer_id, (const void**)&client) == GFALSE)
        {
            return G_ErrNotFound;
        }

        client->on_receive = on_receive;
        client->opaque = opaque;
    }

    return G_OK;
}



#endif